package net.xo68.destiny.core.timingwheel;

import lombok.extern.slf4j.Slf4j;
import net.xo68.destiny.core.timingwheel.impl.AbstractTimingWheel;
import net.xo68.destiny.core.timingwheel.impl.TimingWheelTimeout;

import java.io.Serializable;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

/**
 * 时光轮自旋工作者
 * @author wuxie
 * @version 2020-10-22
 */
@Slf4j
public class TimingWheelWorker<T extends Serializable> implements Runnable {

    private final Set<Timeout<T>> unprocessedTimeouts = new HashSet<>();

    private final TimingWheel<T> timingWheel;

    /**
     * 结束记数器,时光轮调度引擎关闭时，等待关闭调度处理结束再关闭自旋线程池
     */
    private final CountDownLatch shutdowned= new CountDownLatch(1);

    public TimingWheelWorker(TimingWheel<T> timingWheel) {
        this.timingWheel = timingWheel;
    }

    public Set<Timeout<T>> getUnprocessedTimeouts() {
        return unprocessedTimeouts;
    }

    public CountDownLatch getShutdowned() {
        return shutdowned;
    }

    @Override
    public void run() {
        final long workerStartTime=System.currentTimeMillis();
        //计算当前刻度，实际延迟的最终期限
        final long deadline= workerStartTime+ timingWheel.getTickDuration() - timingWheel.getStartTime();
        if(AbstractTimingWheel.workerStateUpdater.get((AbstractTimingWheel<T>) this.timingWheel)== AbstractTimingWheel.WORKER_STATE_STARTED){
            if (deadline > 0){
                long idx = timingWheel.getTick().get() & timingWheel.getMask();
                processCancelledTasks();
                TimingWheelBucket<T> bucket = timingWheel.getWheel().get(idx);
                transferTimeoutsToBuckets();
                bucket.expire(deadline);
                timingWheel.nextTick();
            }
            //log.debug("PendingTimeouts: {}, Buffer.count: {}", timingWheel.getPendingTimeouts().get(), timingWheel.getHashedBuffer().size());
        }
        if(AbstractTimingWheel.workerStateUpdater.get((AbstractTimingWheel<T>) this.timingWheel)== AbstractTimingWheel.WORKER_STATE_SHUTDOWN){
            // Fill the unprocessedTimeouts so we can return them from stop() method.
            for (TimingWheelBucket<T> bucket : timingWheel.getWheel().values()) {
                long removed=bucket.clearTimeouts(unprocessedTimeouts);
                timingWheel.getPendingTimeouts().addAndGet(removed*-1L);
            }
            for (;;) {
                Timeout<T> timeout = timingWheel.getTimeouts().poll();
                if (timeout == null) {
                    break;
                }
                if (!timeout.isCancelled()) {
                    timingWheel.getPendingTimeouts().decrementAndGet();
                    unprocessedTimeouts.add(timeout);
                }
            }
            processCancelledTasks();
            shutdowned.countDown();
        }
    }

    /**
     * 把队列中的超时项转入桶中
     */
    private void transferTimeoutsToBuckets() {
        // transfer only max. 10000 timeouts per tick to prevent a thread to stale the workerThread when it just
        // adds new timeouts in a loop.
        for (int i = 0; i < 10000; i++) {
            Timeout<T> timeout = timingWheel.getTimeouts().poll();
            if (timeout == null) {
                // all processed
                break;
            }
            if (timeout.state() == TimingWheelTimeout.ST_CANCELLED) {
                // Was cancelled in the meantime.
                continue;
            }

            long calculated = timeout.deadline() / timingWheel.getTickDuration();
            timeout.remainingRounds((calculated - timingWheel.getTick().get()) / timingWheel.getTicksPerWheel());

            final long ticks = Math.max(calculated, timingWheel.getTick().get()); // Ensure we don't schedule for past.
            long stopIndex = ticks & timingWheel.getMask();
//            log.debug("tick:{},sourceDeadline:{},deadline:{},要压入桶的任务:{},压入桶的下标:{},轮数:{}",
//                    timingWheel.getTick().get(),
//                    timeout.sourceDeadline(),
//                    timeout.deadline(),
//                    timeout.taskSequence(),
//                    stopIndex,
//                    timeout.remainingRounds());
            TimingWheelBucket<T> bucket = timingWheel.getWheel().get(stopIndex);
            bucket.offer(timeout);
        }
    }

    /**
     * 处理要取消的超时项
     */
    private void processCancelledTasks() {
        for (;;) {
            Timeout<T> timeout = timingWheel.getCancelledTimeouts().poll();
            if (timeout == null) {
                // all processed
                break;
            }
            try {
                timeout.remove();
            } catch (Throwable t) {
                if (log.isWarnEnabled()) {
                    log.warn("An exception was thrown while process a cancellation task", t);
                }
            }
        }
    }

    /**
     * 未处理的超时项
     * @return
     */
    public Set<Timeout<T>> unprocessedTimeouts() {
        return Collections.unmodifiableSet(unprocessedTimeouts);
    }
}
